package com.example.mq.mqserver.core;

import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.VirtualHost;


import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

public class ConsumerManager {
    private VirtualHost parent = null;
    // 指定⼀个线程池，执⾏具体的回调任务
    private ExecutorService workPool = Executors.newFixedThreadPool(4);
    //存放令牌（queueName）的队列
    private BlockingDeque<String> tokenQueue = new LinkedBlockingDeque<>();

    //扫描线程
    private Thread scannerThread = null;
    public ConsumerManager(VirtualHost p) {
        parent = p;

        scannerThread = new Thread( () -> {

            try {
                //1.通过令牌队列拿到queueName
                String queueName = tokenQueue.take();
                //2.根据令牌找到队列
                MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                if (queue == null) {
                  throw new MqException("[ConsumerManager] 取令牌后发现，该队列名不存在 queueName=" + queueName);
                }
                //3.从这个队列中取消息
                synchronized (queue) {
                    consumerMessage(queue);
                }
            } catch (InterruptedException | MqException e) {
                e.printStackTrace();
            }
        });
        //把线程设置为后台线程
        scannerThread.setDaemon(true);
        scannerThread.start();

    }

    //这个方法的调用时机就是发送消息的时候
    public void notifyConsumer(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    public void addConsumer(String consumeTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        //找到对应的队列
       MSGQueue queue =  parent.getMemoryDataCenter().getQueue(queueName);
       if (queue == null) {
           throw new MqException("[ConsumerManager] 队列不存在！ queueName= " + queueName);
       }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumeTag,queueName,autoAck,consumer);
        synchronized (queue) {
            queue.addConsumerEnv(consumerEnv);
            // 如果当前队列中已经有了一些消息了, 需要立即就消费掉.
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < n; i++) {
                // 这个方法调用一次就消费一条消息.
                consumerMessage(queue);
            }
        }
    }

    private void consumerMessage(MSGQueue queue) {
        //1.按照轮询的方式把这个消费者找出来
        ConsumerEnv luckyDog = queue.chooseConsumer();
        if (luckyDog == null) {
            //当前队列没有消费者，暂时不消费，有消费者了再消费
            return ;
        }
        //2.从队列中取出一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null) {
            //此时没有消息，也不需要消费
            return ;
        }
        //3.把消息丢给消费者的回调方法中，丢给线程池
        workPool.submit(() -> {
            try {
                //1.把消息先放在待确定消息队列
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(),message);
                //2.真正执行回调操作
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumeTag(), message.getBasicProperties(), message.getBody());
                //3.
                if (luckyDog.isAutoAck()) {
                    //4.先删除硬盘上的消息，判断是否持久化
                    if (message.getDeliverMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(queue,message);
                    }
                    //5.删除内存上的消息
                    parent.getMemoryDataCenter().deleteMessage(message.getMessageId());
                    //6.删除待确认消息队列中的消息
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId());
                    System.out.println("[ConsumerManager] 消息被成功消费 queueName=" + queue.getName());
                }

            } catch (Exception e) {
                e.printStackTrace();
            }

        });
    }
}
